
#include "as_acceptor.h"

#include <signal.h>
#include <iostream>

#ifdef _WIN32
#include <winsock2.h>
#include <ioapiset.h>
#include <MSWSock.h>
#include <winnt.h>

//tcpkeeplive
#include <mstcpip.h>
#else
#include <sys/epoll.h>
#endif

#include "as_socket.h"
#include "net_addr.h"
#include "as_session.h"
#include "as_session_mgr.h"
#include "byte_buffer.h"

#ifdef _WIN32
LPFN_GETACCEPTEXSOCKADDRS as::inet::AsAcceptor::m_accept_addr_fun = nullptr;
LPFN_ACCEPTEX as::inet::AsAcceptor::m_accept_fun = nullptr;
#endif

std::map<int, std::function<void()> > as::inet::AsAcceptor::m_signal_map;

namespace as
{
namespace inet
{

AsAcceptor::AsAcceptor()
{
#ifdef _WIN32
    WSADATA data;
    WORD wVersion = MAKEWORD(2, 2);
    int err = WSAStartup(wVersion, &data);
    assert(err == 0);
#endif
    m_listenSocket = std::make_unique<as::inet::AsSession>();
    m_eventFlags = false;

#ifdef _WIN32
    m_ioPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0);
#else
    m_ioPort = epoll_create1(EPOLL_CLOEXEC);
#endif
}

AsAcceptor::~AsAcceptor()
{
}

AsAcceptor* AsAcceptor::set_workthread_num(int num)
{
    m_workThreadNum = num;
    return this;
}

AsAcceptor* AsAcceptor::start(as::inet::NetAddr &addr)
{
    m_state = io_state::RUNNING;

    m_listenSocket->init();
    if(m_listenSocket->listen(addr) == nullptr)
    {
        m_listenSocket->close();
        //LOG_ERROR << "acceptor start NetAddr listen error" << std::endl;
        return this;
    }

    socket_fd fd = m_listenSocket->get_socket_fd();
    //set socket keep alive and non block
    as::inet::AsSocket::set_keep_alive(fd);
    as::inet::AsSocket::set_non_block(fd);

    register_sys_fun();
    create_io_and_post_event();

    for(int i = 0; i < m_workThreadNum; i++)
    {
        m_workThreads.emplace_back(std::thread(&AsAcceptor::work_run, this));
    }
    return this;
}

AsAcceptor* AsAcceptor::start(std::string_view ip)
{
    return this;
}

AsAcceptor* AsAcceptor::start(std::string_view host, int port)
{
    m_state = io_state::RUNNING;

    as::inet::NetAddr addr(host, port);
    
    m_listenSocket->init();
    
    if(m_listenSocket->listen(addr) == nullptr)
    {
        m_listenSocket->close();
        //LOG_ERROR << "acceptor start NetAddr listen error" << std::endl;
        return this;
    }

    socket_fd fd = m_listenSocket->get_socket_fd();
    //set socket keep alive and non block
    as::inet::AsSocket::set_keep_alive(fd);
    as::inet::AsSocket::set_non_block(fd);

    register_sys_fun();
    create_io_and_post_event();

    for(int i = 0; i < m_workThreadNum; i++)
    {
        m_workThreads.emplace_back(std::thread(&AsAcceptor::work_run, this));
    }
    return this;
}

void AsAcceptor::register_sys_fun()
{
#ifdef _WIN32
    GUID guidAcceptEx = WSAID_ACCEPTEX;
	GUID guidGetAcceptSockAddrs = WSAID_GETACCEPTEXSOCKADDRS;
	DWORD dwBytes = 0;
	ASSERT(SOCKET_ERROR != WSAIoctl(
        m_listenSocket->get_socket_fd(),
		SIO_GET_EXTENSION_FUNCTION_POINTER,
		&guidAcceptEx,
		sizeof(guidAcceptEx),
		&m_accept_fun,
		sizeof(m_accept_fun),
		&dwBytes,
		NULL,
		NULL));

    ASSERT(SOCKET_ERROR != WSAIoctl(
		m_listenSocket->get_socket_fd(),
		SIO_GET_EXTENSION_FUNCTION_POINTER,
		&guidGetAcceptSockAddrs,
		sizeof(guidGetAcceptSockAddrs),
		&m_accept_addr_fun,
		sizeof(m_accept_addr_fun),
		&dwBytes,
		NULL,
		NULL));
#endif
}

void AsAcceptor::create_io_and_post_event()
{
#ifdef _WIN32
    SYSTEM_INFO SystemInfo;
    GetSystemInfo(&SystemInfo);

    register_sys_fun();

	if (CreateIoCompletionPort((HANDLE)m_listenSocket->get_socket_fd(), m_ioPort, 0, 0) == nullptr)
    //if(CreateIoCompletionPort((HANDLE)m_listenSocket->get_fd(), m_ioPort, (ULONG_PTR)&m_listenSocket, 0) == nullptr)
    {
        m_listenSocket->close();
        ASSERT(nullptr);
    }


    for(auto &io : m_events)
    {
        if(post_accept(&io) == false)
        {
            io.clear();
        } 
    }
    m_eventFlags = true;

#else
    IOContext *listenIo = m_listenSocket->pop_io_context();
    listenIo->socket = m_listenSocket->get_socket_fd();
    listenIo->type = IOType::Read;

    LOG_INFO << "listen fd:" <<listenIo->socket << std::endl;
    post_accept(listenIo); 
#endif
}

void AsAcceptor::stop()
{
    m_state = io_state::STOP;

    m_listenSocket->close();
    for(auto &thread : m_workThreads)
    {
        //auto &thread = m_workThreads[i];
        while(thread.joinable() == false)
        {
			std::this_thread::sleep_for(std::chrono::milliseconds(300));
        }

        thread.join();
    }

    m_state = io_state::STOP_OK;
}

bool AsAcceptor::post_accept(IOContext *io)
{
#ifdef _WIN32
    io->type = IOType::Accept;
    io->socket = WSASocketW(AF_INET, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
    if(io->socket == SOCKET_ERR_FD)
    {
        return false;
    }
	DWORD dwBytes = 0;

    bool ret = m_accept_fun(m_listenSocket->get_socket_fd(), io->socket, io->wsaBuf.buf, 0, sizeof(sockaddr_in) + 16, sizeof(sockaddr_in) + 16, &dwBytes, &io->overlapped);
    if(ret == false)
    {
        auto err = GetLastError();
        //LOG_INFO << "post accept error:" << err << std::endl;
        if(WSA_IO_PENDING != err)
        {
            return false;
        }
    }

    if(setsockopt(io->socket, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char*)&m_listenSocket, sizeof(m_listenSocket)) == false)
    {
        auto err = GetLastError();
        //LOG_INFO << "set post accept opt error:" << err << std::endl;
        return false;
    }

    //LOG_INFO << "pose accept socket fd:" << io->socket << std::endl;
#else
    static epoll_event ev;
    memset(&ev, 0, sizeof(ev));
    ev.data.ptr = io;
    ev.events = io->type;
    int ret = -1;
    ret = ::epoll_ctl(m_ioPort, EPOLL_CTL_ADD, io->socket, &ev);
    if(ret < 0)
    {
        LOG_ERROR << "epoll post accept error:" << ret << std::endl;
        return false;
    }
#endif
    return true;
}

bool AsAcceptor::post_recv(IOContext *io)
{
#ifdef _WIN32
    DWORD dwFlags = 0, dwBytes = 0;
	io->clear();
	io->type = IOType::Read;

	int nBytesRecv = WSARecv(io->socket, &io->wsaBuf, 1, &dwBytes, &dwFlags, &io->overlapped, NULL);
	if (SOCKET_ERROR == nBytesRecv)
	{
        int err = WSAGetLastError();
        if(WSA_IO_PENDING != err)
        {
            //LOG_INFO << "post recv err:" << err << std::endl;
            as::inet::AsSocket::close(io->socket);
            auto tempSession = io->ioSession;
            io->ioSession->push_io_context(io);
            return false;
        }
	}
#else
    static epoll_event ev;
    memset(&ev, 0, sizeof(ev));
    ev.data.ptr = io;
    ev.events = io->type;
    int ret = ::epoll_ctl(m_ioPort, EPOLL_CTL_MOD, io->socket, &ev);
    if(ret < 0)
    {
        LOG_ERROR << "epoll pose recv error:" << ret << std::endl;
        return false;
    }
#endif
    return true;
}

bool AsAcceptor::post_write(IOContext *io)
{
    return true;
}

#ifdef _WIN32
int AsAcceptor::DoWinAccept(IOContext *io)
{
	SOCKADDR_IN *clientAddr = NULL;
	SOCKADDR_IN *localAddr = NULL;
	int clientAddrLen, localAddrLen;
	clientAddrLen = localAddrLen = sizeof(SOCKADDR_IN);

	m_accept_addr_fun(io->wsaBuf.buf, 0, localAddrLen, clientAddrLen, (LPSOCKADDR *)&localAddr, &localAddrLen, (LPSOCKADDR *)&clientAddr, &clientAddrLen);
    
    socket_fd clientfd = io->socket;
    //repeat post new accept io
    io->clear();
    if(post_accept(io) == false)
    {
        io->clear();
        //LOG_INFO << "Do win accept error" << std::endl;
    }

    if (CreateIoCompletionPort((HANDLE)clientfd, m_ioPort, NULL, 0) == nullptr)
    {
        DWORD dwErr = WSAGetLastError();
        if (dwErr != ERROR_INVALID_PARAMETER)
        {
            //LOG_INFO << "session fd:" << clientfd << " listen fd:" << m_listenSocket->get_socket_fd() << std::endl;
            //LOG_INFO << "create completion port error:" << dwErr << std::endl;
            as::inet::AsSocket::close(clientfd);
            return -1;
        }
    }

    //create new io post recv event
    std::shared_ptr<as::inet::AsSession> ioSession = std::make_shared<as::inet::AsSession>();
    auto newIOContext = ioSession->pop_io_context();
    newIOContext->type = IOType::Read;
    newIOContext->socket = clientfd;
    newIOContext->ioSession = ioSession;
    ioSession->init(clientfd);

    if(post_recv(newIOContext) == false)
    {
        return -1;
    }

    if(m_new_con_cb)
    {
        m_new_con_cb(ioSession);
    }
	return 0;
}

#else
int AsAcceptor::DoUnixAccept(IOContext *io)
{
    struct sockaddr_in client_addr;
    memset(&client_addr, 0, sizeof(client_addr));
    socklen_t clientaddrlen = 0;
    int fd = accept4(m_listenSocket->get_socket_fd(), (struct sockaddr *)&client_addr, &clientaddrlen, SOCK_NONBLOCK);
    if(fd == SOCKET_ERR_FD)
    {
        LOG_ERROR << "do unix accept error fd:" << fd << " errorId:" << errno << " msg:" << strerror(errno) << std::endl;
        return -1;
    }

    auto session = std::make_shared<as::inet::AsSession>();
    session->init(fd)
           ->set_client_addr(&client_addr);

    if(m_new_con_cb)
    {
        m_new_con_cb(session);
    }

    auto newio = session->pop_io_context();
    newio->type = IOType::Read;
    newio->socket = session->get_socket_fd();
    newio->ioSession = session;
    static epoll_event ev;
    memset(&ev, 0, sizeof(ev));
    ev.data.ptr = newio;
    ev.events = newio->type;
    int ret = ::epoll_ctl(m_ioPort, EPOLL_CTL_ADD, newio->socket, &ev);
    if(ret < 0)
    {
        LOG_ERROR << "epoll do accept error:" << ret << std::endl;
        return -2;
    }
    return 0;
}

#endif

void AsAcceptor::work_run()
{
#ifdef _WIN32
    OVERLAPPED_ENTRY overlappeds[SVC_MAX_EVENT_NUM]; 
#endif
    while (m_state == io_state::RUNNING)
    {
#ifdef _WIN32
        ULONG eventNum = 0;
        if(!GetQueuedCompletionStatusEx(m_ioPort, &overlappeds[0], SVC_MAX_EVENT_NUM, &eventNum, INFINITE, false))
        {
            auto err = WSAGetLastError();
            if(WAIT_TIMEOUT != err)
            {
                //LOG_INFO << "error" << std::endl;
            }

            as::tool::sleepms(300);
            continue;
        }

        if(m_state != io_state::RUNNING)
        {
            break;
        }

        if(eventNum > 0)
        {
            for(int i = 0; i < (int)eventNum; ++i)
            {
                auto &event = overlappeds[i];
                //printf("work run iocontext addr:%x", &event);
				LPOVERLAPPED &ol = event.lpOverlapped;
				if (ol == 0) {
					std::cout << ("[iocp]GetQueuedCompletionStatusEx, no packet dequeue");
					return;
				}

                DWORD transNum = event.dwNumberOfBytesTransferred;
                IOContext *io= (CONTAINING_RECORD(ol, IOContext, overlapped));
                if(io)
                {
                    std::cout << "idx:" << i << " type:" << io->type << " num:" << transNum << std::endl;
                    switch(io->type)
                    {
                        case IOType::Accept:
                        {
                            DoAccept(io);
                            break;
                        }
                        case IOType::Read:
                        {
                            if (transNum <= 0)
                            {
                                push_io_context(io);
                                break;
                            }
                            DoRecv(io, (int)transNum);
                            break;
                        }
                        case IOType::Write:
                        {
                            if(transNum <= 0)
                            {
                                push_io_context(io);
                                break;
                            }
                            DoWrite(io, transNum);
                            break;
                        }
                    }
                }
            }
        }
#else
        int n = ::epoll_wait(m_ioPort, &m_events[0], SVC_MAX_EVENT_NUM, -1); 
        if(n < 0)
        {
            LOG_ERROR << "epoll wait empty event:" << n <<  std::endl;
            return;
        }

        for(int i = 0; i < n; ++i)
        {
            IOContext *io = (IOContext*)m_events[i].data.ptr;
            if(io)
            {
                LOG_INFO << "event fd:" << io->socket << " listen socket fd:" << m_listenSocket->get_socket_fd() << std::endl;
                if(io->socket == m_listenSocket->get_socket_fd())
                {
                    DoAccept(io);
                }
                else
                {
                    auto clientSession = io->ioSession;
                    if(clientSession)
                    {
                        LOG_INFO << "io type:" << io->type << " enum:" << IOType::Read << std::endl;
                        if(io->type == IOType::Read)
                        {
                            int curPos = 0;
                            int ret = 0;
                            auto tmp = &io->buffer[0];
                            do
                            {
                                ret = AsSocket::recv(clientSession->get_socket_fd(), tmp, IO_CONTEXT_BUF_SIZE);
                                if(ret > 0)
                                {
                                    curPos += ret;
                                    tmp = &io->buffer[curPos];
                                    LOG_INFO << "recv again:" << ret << std::endl;
                                }
                                else if (ret < 0)
                                {
                                    LOG_INFO << "recv wait ..." << std::endl;
                                    SLEEP_MS(20);
                                }
                            } while (ret == 0);

                            if(m_recv_cb)
                            {
                                m_recv_cb(io->ioSession, io->buffer, curPos);
                            }
                        }
                    }
                }
            }     
        }
#endif
    }


    //LOG_INFO << "work thread break:" << std::this_thread::get_id() << std::endl;
}

void AsAcceptor::run()
{
    while (m_state != io_state::STOP_OK)
    {
#ifdef _WIN32
#else
        
#endif
    }
}

void AsAcceptor::signal_fun(int sig)
{
    m_signal_map[sig]();
}

void AsAcceptor::RegSignal(int sig, const std::function<void()> cb_fun)
{
    m_signal_map[sig] = cb_fun;
    //::signal(sig, signal_fun);
}

int AsAcceptor::DoAccept(IOContext *io)
{
#ifdef _WIN32
    DoWinAccept(io);
#else
    DoUnixAccept(io);
#endif
	
    return 0;
}

int AsAcceptor::DoRecv(IOContext *io, int byteNum)
{
    if(m_recv_cb)
    {
        m_recv_cb(io->ioSession, (char*)io->buffer, byteNum);
    }

    io->clear();
#ifdef _WIN32
    if(post_recv(io) == false)
    {
        as::inet::AsSocket::close(io->socket);
        io->ioSession->push_io_context(io);
        return -1;
    }

#endif
    return 0;
}

int AsAcceptor::DoWrite(IOContext *io, int byteNum)
{
    if(m_write_cb)
    {
        m_write_cb(io->ioSession, byteNum);
    } 

    return 0;
}

IOContext* AsAcceptor::pop_io_context(socket_fd fd)
{
    auto io = m_listenSocket->pop_io_context();
    io->socket = fd;
    return io;
}

void AsAcceptor::push_io_context(IOContext *io)
{
    if(m_close_cb)
    {
        m_close_cb(io->ioSession);
    }

    m_listenSocket->push_io_context(io);
}




}


}